<?php

namespace Yunque\Client\driver;

use Kafka\Producer;
use Kafka\ProducerConfig;
use Yunque\Client\utils\Bytes;

class Kafka implements Driver
{
    public Producer $producer;

    public static function init($config)
    {
        $self = new self();
        $producerConfig = \Kafka\ProducerConfig::getInstance();
        $producerConfig->setMetadataBrokerList($config['brokerList']);
        /* Topic的元信息刷新的间隔 */
        $producerConfig->setMetadataRefreshIntervalMs($config["metadataRefreshIntervalMs"]);
//        $config->setGroupId($config['groupId']);
        /* 设置broker的代理版本 */
        $producerConfig->setBrokerVersion($config['brokerVersion']);
        /* 只需要leader确认消息 */
        $producerConfig->setRequiredAck($config['requiredAck']);
        /* 是否异步 */
        $producerConfig->setIsAsyn($config['isAsyn']);
        /* 每500毫秒发送消息 */
        $producerConfig->setProduceInterval($config['produceInterval']);
        /* 创建⼀个⽣产者实例 */
        $self->producer = new Producer();
        return $self;
    }

//    public static function init(array $config = null): ?Kafka
//    {
//        if ($config==null){
//            return null;
//        }
//        $self=new self();
//
//        $producerConfig = ProducerConfig::getInstance();
//
////        print_r($config['host']);
//        $producerConfig->setMetadataBrokerList($config['host'].':'.$config['port']);
//        /* Topic的元信息刷新的间隔 */
//        $producerConfig->setMetadataRefreshIntervalMs($config["metadataRefreshIntervalMs"]);
////        $config->setGroupId($config['groupId']);
//        /* 设置broker的代理版本 */
//        $producerConfig->setBrokerVersion($config['brokerVersion']);
//        /* 只需要leader确认消息 */
//        $producerConfig->setRequiredAck(1);
//        /* 是否异步 */
//        $producerConfig->setIsAsyn($config['isAsyn']);
//        /* 每500毫秒发送消息 */
//        $producerConfig->setProduceInterval(500);
//        /* 创建⼀个⽣产者实例 */
//        $self->producer = new Producer();
//        return $self;
//    }
    //发消息
    //发消息
    public function sendTo($topic, $msg)
    {
//        $producer = new \Kafka\Producer();
        $result =  $this->producer->send([[
                'topic' => $topic,
                'value' => $msg,
                'key' => $topic]]
        );
        print_r($result);
    }

}